// author: shelman
// description：封装了简单的线程池

#pragma once

#include <iostream>
#include <algorithm>
#include <iterator>
#include <string>
#include <mutex>
#include <thread>
#include <condition_variable>
#include <unordered_map>
#include <list>
#include <functional>
#include <memory>
#include <atomic>
#include <deque>
#include <queue>
#include <utility>


template<typename T>
class SyncQueue
{
public:
    SyncQueue(uint32_t ui32MaxSize)
        : m_bStop(false)
        , m_ui32MaxSize(ui32MaxSize)
    {

    }

    ~SyncQueue()
    {
        stop();
    }

    void put(const T& x)
    {
        add(x);
        return;
    }

    void put(T&& x)
    {
        add(std::forward<T>(x));
        return;
    }

    void take(std::list<T>& list)
    {
        std::unique_lock<std::mutex> locker(m_mutex);
        m_cvNotEmpty.wait(locker, [this] {return (m_bStop || notEmpty());});

        if (m_bStop)
        {
            return;
        }

        list = std::move(m_lstQueue);
        m_cvNotFull.notify_one();

        return;
    }

    void take(T& t)
    {
        std::unique_lock<std::mutex> locker(m_mutex);
        m_cvNotEmpty.wait(locker, [this] {return (m_bStop || notEmpty());});

        if (m_bStop)
        {
            return;
        }

        t = m_lstQueue.front();
        m_lstQueue.pop_front();
        m_cvNotFull.notify_one();

        return;
    }

    void stop()
    {
        {
            std::lock_guard<std::mutex> locker(m_mutex);
            m_bStop = true;
        }

        m_cvNotFull.notify_all();
        m_cvNotEmpty.notify_all();

        return;
    }

    bool empty()
    {
        std::lock_guard<std::mutex> locker(m_mutex);
        return m_lstQueue.empty();
    }

    bool full()
    {
        std::lock_guard<std::mutex> locker(m_mutex);
        return (m_lstQueue.size() == m_ui32MaxSize);
    }

    size_t size()
    {
        std::lock_guard<std::mutex> locker(m_mutex);
        return m_lstQueue.size();
    }

    int count()
    {
        return size();
    }

private:
    bool notFull() const
    {
        bool full = m_lstQueue.size() >= m_ui32MaxSize;
        return !full;
    }

    bool notEmpty() const
    {
        bool empty = m_lstQueue.empty();
        return !empty;
    }

    void add(T&& x)
    {
        std::unique_lock< std::mutex> locker(m_mutex);
        m_cvNotFull.wait(locker, [this] {return (m_bStop || notFull());});

        if (m_bStop)
        {
            return;
        }

        m_lstQueue.emplace_back(std::forward<T>(x));
        m_cvNotEmpty.notify_one();

        return;
    }

    void add(const T& x)
    {
        std::unique_lock< std::mutex> locker(m_mutex);
        m_cvNotFull.wait(locker, [this] {return (m_bStop || notFull());});

        if (m_bStop)
        {
            return;
        }

        m_lstQueue.emplace_back(x);
        m_cvNotEmpty.notify_one();

        return;
    }

private:
    bool m_bStop; // 停止的标志
    uint32_t m_ui32MaxSize; // 同步队列最大的size
    std::mutex m_mutex; // 互斥量和条件变量结合起来使用
    std::condition_variable m_cvNotEmpty; // 不为空的条件变量
    std::condition_variable m_cvNotFull;  // 没有满的条件变量
    std::list<T> m_lstQueue; // 缓冲区
};


class ThreadPool
{
public:
    using Task = std::function<void()>; // 建议运行的Task中没有一直为真的循环执行，具体执行循环逻辑由ThreadPool管理

    ThreadPool(uint32_t ui32NumThreads = std::thread::hardware_concurrency())
        : m_objQueue(ui32NumThreads)
        , m_ui32ThreadSize(ui32NumThreads)
    {

    }

    ~ThreadPool()
    {
        stop(); // 如果没有停止时则主动停止线程池
    }

    void addTask(Task&& task)
    {
        m_objQueue.put(std::forward<Task>(task));
    }

    void addTask(const Task& task)
    {
        m_objQueue.put(task);
    }

    void start(bool bLoopIn = true)
    {
        m_bRunning = true;

        //创建线程组
        for (uint32_t i = 0; i < m_ui32ThreadSize; ++i)
        {
            if (bLoopIn)
            {
                m_lst4ThreadGroup.emplace_back(std::make_shared<std::thread>(&ThreadPool::run, this));
            }
            else
            {
                m_lst4ThreadGroup.emplace_back(std::make_shared<std::thread>(&ThreadPool::run_ex, this));
            }
        }

        return;
    }

    void stop()
    {
        std::call_once(m_stFlag, [this] { stopThreadGroup(); }); // 保证多线程情况下只调用一次
    }

private:
    void run()
    {
        while (m_bRunning)
        {
            // Task中不建议执行循环业务等
            // 取任务分别执行
            Task task;
            m_objQueue.take(task);

            try
            {
                if (task)
                {
                    task();
                }
            }
            catch (std::exception& e)
            {
                std::cout << "ThreadPool::run, WARNING: running task, with exception, " \
                          << e.what() << std::endl;
                return;
            }
        }

        return;
    }

    void run_ex()
    {
        // Task中执行循环业务等
        // 取任务分别执行
        Task task;
        m_objQueue.take(task);

        try
        {
            if (task)
            {
                task();
            }
        }
        catch (std::exception& e)
        {
            std::cout << "ThreadPool::run, WARNING: running task, with exception, " \
                      << e.what() << std::endl;
        }

        return;
    }

    void stopThreadGroup()
    {
        m_objQueue.stop(); // 让同步队列中的线程停止
        m_bRunning = false; // 置为false，让内部线程跳出循环并退出

        for (auto thread : m_lst4ThreadGroup) // 等待线程结束
        {
            if (thread->joinable())
            {
                thread->join();
            }
        }
        m_lst4ThreadGroup.clear();
    }

private:
    std::list<std::shared_ptr<std::thread> > m_lst4ThreadGroup; // 处理任务的线程组
    SyncQueue<Task> m_objQueue; // 同步队列
    std::atomic_bool m_bRunning; // 是否停止的标志
    std::once_flag m_stFlag; // 标志
    uint32_t m_ui32ThreadSize; // 线程个数
};
